package com.lsx143.wordcount.day3.sink;

import com.lsx143.wordcount.bean.WaterSensor;
import com.lsx143.wordcount.day2.RandomWaterSensor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.PreparedStatement;
import java.sql.SQLException;

public class MyJDBCSink {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env
                .addSource(new RandomWaterSensor(RandomWaterSensor.RUN_MODE.RANDOM))
                .addSink(
                        JdbcSink.sink(
                                "insert into sensor values(?,?,?)",
                                new JdbcStatementBuilder<WaterSensor>() {
                                    @Override
                                    public void accept(PreparedStatement ps, WaterSensor waterSensor) throws SQLException {
                                        ps.setString(1, waterSensor.getId());
                                        ps.setLong(2, waterSensor.getTs());
                                        ps.setInt(3, waterSensor.getVc());
                                    }
                                },
                                JdbcExecutionOptions
                                        .builder()
                                        .withMaxRetries(5)
                                        .withBatchSize(1000)
                                        .withBatchIntervalMs(200)
                                        .build(),
                                new JdbcConnectionOptions
                                        .JdbcConnectionOptionsBuilder()
                                        .withUrl("jdbc:mysql://hadoop102:3306/test?useSSL=false")
                                        .withUsername("root")
                                        .withPassword("123321")
                                        .build()
                        )
                );
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
